package ru.unlocker.reports.domain;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public abstract class CalculateReportStatusJob {

  public final StreamExecutionEnvironment createApplicationPipeline(
      ParameterTool params, StreamExecutionEnvironment env
  ) {
    env.getConfig().setGlobalJobParameters(params);
    var initialStream = readStatusStream(params, env);
    var statusStream = initialStream
        .keyBy(ReportAndTimerEvent::getKey)
        .process(new StatusProcessor());
    writeStatusStream(params, statusStream);
    return env;
  }

  protected abstract DataStream<ReportAndTimerEvent> readStatusStream(
      ParameterTool params, StreamExecutionEnvironment env
  );

  protected abstract void writeStatusStream(ParameterTool params, DataStream<ReportStatus> outStream);
}
